package com.shujia.spark.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.io.Source

object Demo23Broadcast {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()

    conf.setMaster("local")

    conf.setAppName("wc")

    val sc = new SparkContext(conf)

    /**
     * 广播变量
     * 当在算子内使用算子外的一个比较大的变量时，可以将这个变量广播出去，可以减少变量的副本数
     *
     */


    //读取学生表，以学号作为key构建一个map集合
    val studentMap: Map[String, String] = Source
      .fromFile("data/students.txt")
      .getLines()
      .toList
      .map(stu => {
        val ud: String = stu.split(",")(0)
        (ud, stu)
      })
      .toMap


    val scoresRDD: RDD[String] = sc.textFile("data/score.txt", 10)

    println(s"scoresRDD：${scoresRDD.getNumPartitions}")

    /**
     * 关联学生表和分数表
     * 循环分数表，使用学号到学生表的mao集合中查询学生的信息
     *
     */

    /**
     * 将Driver端的一个普通变量广播到Executor端
     *
     */
    val broadCastMap: Broadcast[Map[String, String]] = sc.broadcast(studentMap)

    val joinRDD: RDD[(String, String)] = scoresRDD.map(sco => {
      val id: String = sco.split(",")(0)

      //使用学号到学生表中获取学生的信息

      /**
       * 在算子内使用广播变量
       * 1、当第一个task在执行过程中如果使用了广播变量，会向Executor获取广播变量
       * 2、如果Executor中没有这个广播变量，Executor会去Driver端获取
       * 3、如果下一个task再使用到这个广播变量就可以直接用了
       *
       */
      //在算子内获取广播变量
      val map: Map[String, String] = broadCastMap.value
      val studentInfo: String = map.getOrElse(id, "默认值")

      (sco, studentInfo)
    })


    joinRDD.foreach(println)


  }

}
